package net.bwie.order.dwd.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import net.bwie.realtime.guanjuntao.util.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 * 异步请求HBase，从维表获取数据
 * @author xuanyu
 * @date 2025/2/8
 */
public class RequestDicDimAsyncFunction extends RichAsyncFunction<String, String> {

    /**
     * 线程池操作对象
     */
    private ExecutorService executorService;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建线程池
        executorService = Executors.newScheduledThreadPool(10) ;
    }

    @Override
    public void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {
        // a-解析JSON字符串
        JSONObject jsonObject = JSON.parseObject(input);
        String rowKey = jsonObject.getString("source_type");

        // b-异步请求HBase获取数据
        final Future<String> result = executorService.submit(
                new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        return HbaseUtil.getData("dim_base_dic", rowKey, "info", "dic_name");
                    }
                }
        ) ;

        // c-异步返回结果
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    String requestValue = result.get();
                    // 添加字段值
                    jsonObject.put("source_type_name", requestValue);
                } catch (InterruptedException | ExecutionException e) {
                    // 设置字段为：exception
                    jsonObject.put("source_type_name", "exception");
                }
                // 返回值
                return jsonObject.toString() ;
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(dbResult));
        });

    }


    @Override
    public void timeout(String input, ResultFuture<String> resultFuture) throws Exception {
        // 解析JSON
        JSONObject jsonObject = JSON.parseObject(input);
        // 设置字段为：timeout
        jsonObject.put("source_type_name", "timeout");
        // 返回
        resultFuture.complete(Collections.singleton(jsonObject.toString()));
    }


    @Override
    public void close() throws Exception {
        executorService.shutdownNow();
    }

}
